package f6sink;

import flinkemp.Emp;
import flinkemp.EmpFun;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.connector.jdbc.JdbcSink;

import java.sql.PreparedStatement;
import java.sql.SQLException;


public class MysqlSink {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Emp> data = env.addSource(new EmpFun());
        data.print();

        data.addSink(JdbcSink.sink(
                "INSERT INTO emp (empNo,eName,job,sal,deptNo) VALUES (?,?,?,?,?)",
                new JdbcStatementBuilder<Emp>() {

                    @Override
                    public void accept(PreparedStatement preparedStatement, Emp emp) throws SQLException {
                        preparedStatement.setInt(1, emp.empNo);
                        preparedStatement.setString(2, emp.eName);
                        preparedStatement.setString(3, emp.job);
                        preparedStatement.setDouble(4, emp.sal);
                        preparedStatement.setInt(5, emp.deptNo);
                    }
                },
                JdbcExecutionOptions.builder()
                        .withBatchSize(5)
                        .withBatchIntervalMs(200)
                        .withMaxRetries(2)
                        .build(),

                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:mysql://localhost:3306/testddl")
                        .withDriverName("com.mysql.jdbc.Driver")
                        .withUsername("root")
                        .withPassword("123456")
                        .build()
        ));
        env.execute();
    }
}
